set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode='nonstrict';
set hive.exec.max.dynamic.partitions=400;
set hive.exec.max.dynamic.partitions.pernode=400;

insert overwrite  table jms_dm.dm_false_send_xyf_detail_dt
select
     base.waybill_no                   as waybill_no                 --运单编号
    ,step.ordersource_code             as ordersource_code          --订单来源编码
    ,step.ordersource_name             as ordersource_name          --订单来源名称
    ,dim.code                          as start_network_code        --始发网点编码
    ,dim.name                          as start_network_name        --始发网点名称
    ,dim.agent_code                    as start_agent_code          --始发代理区编码
    ,dim.agent_name                    as start_agent_name          --始发代理区名称
    ,nvl(dim.virt_code,dim.agent_code) as start_virt_code           --始发虚拟代理区编码
    ,nvl(dim.virt_name,dim.agent_name) as start_virt_name           --始发虚拟代理区名称
    ,dim.provider_id                   as start_provider_id         --始发省份编码
    ,dim.provider_desc                 as start_provider_desc       --始发省份名称
    ,dim.city_id                       as start_city_id             --始发城市编码
    ,dim.city_desc                     as start_city_desc           --始发城市名称
    ,dim_start.code                    as start_center_code         --始发中心编码
    ,dim_start.name                    as start_center_name         --始发中心名称
    ,step.taking_time                  as taking_time               --揽收时间
    ,step.network_send_time            as network_send_time         --始发网点发件时间
    ,wide.start_center_arrival_time    as start_center_arrival_time --始发中心到件时间
    ,wide.start_center_send_time       as start_center_send_time    --始发中心发件时间
    ,wide.customer_order_time          as customer_order_time       --客户下单时间
    ,base.is_upload_flag               as is_upload_flag            --是否手动补录
    ,base.upload_batch                 as upload_batch              --批次号
    ,date_sub('{{ execution_date | cst_ds }}',1)          as dt                        --分区日期
from (
    select
         waybill_no
        ,max(is_upload_flag) as is_upload_flag
        ,max(upload_batch) as upload_batch
    from (
        select
             taking.waybill_no
            ,0 as is_upload_flag
            ,null as upload_batch
        from (
            select *
            from (
                select
                      waybill_no
                     ,scan_time as taking_time
                     ,row_number() over(partition by waybill_no,scan_site_code order by scan_time) as rn
                from jms_dwd.dwd_tab_barscan_taking_base_dt --揽收
                where dt between date_sub('{{ execution_date | cst_ds }}',1) and '{{ execution_date | cst_ds }}'
                and to_date(scan_time) =  date_sub('{{ execution_date | cst_ds }}',1)
            ) a where a.rn = 1
        ) taking
        left join (
            select
                  waybill_no
            from jms_dwd.dwd_wide_abnormal_union_waybill_dt
            where dt between date_sub('{{ execution_date | cst_ds }}',45) and '{{ execution_date | cst_ds }}'
            and scan_type in ('reback','intercept')
            group by waybill_no
        ) abnormal on taking.waybill_no = abnormal.waybill_no
        where abnormal.waybill_no is null --剔除退件、转寄件、拦截件
        union all
        select
             waybill_no
            ,1 as is_upload_flag
            ,upload_batch
        from (
            select
                 bill_no as waybill_no
                ,upload_batch
                ,row_number() over (partition by bill_no order by upload_batch desc ) as rn
            from jms_dwd.dwd_tab_false_ship_upload_det_dt --平台上传运单
            where dt = '{{ execution_date | cst_ds }}'
            and platform = '逍遥峰'
        ) upload where upload.rn = 1
    ) a group by waybill_no  --算出单号
) base
inner join (
    select
        step.*
    from (
        select
             waybill_no                      as waybill_no
            ,send_station_network_code       as scan_site_code
            ,send_station_network_type       as scan_site_type
            ,send_station_number             as scan_site_number
            ,send_station_send_later_time    as scan_site_send_time
            ,send_station_arrival_early_time as scan_site_arrival_time
            ,send_center_station_number      as send_center_station_number
            ,taking_early_scantime           as taking_time
            ,send_station_early_scan_time    as scan_site_min_time
            ,next_station_early_scan_time    as next_site_min_time
            ,send_station_send_later_time    as network_send_time
            ,ordersource_code                as ordersource_code
            ,ordersource_name                as ordersource_name
        from jms_dwd.dwd_wide_unsign_trace_waybill_step_dt
        where dt between date_sub('{{ execution_date | cst_ds }}',1) and '{{ execution_date | cst_ds }}'
        and to_date(taking_early_scantime) = date_sub('{{ execution_date | cst_ds }}',1)
        and ordersource_name in ('逍遥峰')
        and !(send_station_number = 1 and coalesce(instore_early_scantime,outstore_early_scantime,sign_early_scantime) is not null) --剔除自收自派
        and send_station_number = 1 
    ) step 
    left join (
        select *
        from (
            select
                  waybill_no
                 ,scan_time as send_time
                 ,scan_site_code
                 ,row_number() over(partition by waybill_no,scan_site_code order by scan_time) as rn
            from jms_dwd.dwd_tab_barscan_send_base_dt --发件
            where dt between date_sub('{{ execution_date | cst_ds }}',1) and '{{ execution_date | cst_ds }}'
        ) a where a.rn = 1
    ) send on step.waybill_no = send.waybill_no 
          and step.scan_site_code = send.scan_site_code 
    where (unix_timestamp(coalesce(if(send_time > next_site_min_time,next_site_min_time,send_time),next_site_min_time,concat('{{ execution_date | cst_ds }}',' 23:59:59'))) - unix_timestamp(taking_time)) /60 /60 > 24
) step on base.waybill_no = step.waybill_no
left join (
    select
         waybill_no                      as waybill_no
        ,pick_network_code               as pick_network_code
        ,customer_order_time             as customer_order_time       --客户下单时间
        ,first_center_arrival_early_time as start_center_arrival_time --始发中心到件时间
        ,first_center_send_later_time    as start_center_send_time    --始发中心发件时间
        ,first_center_network_code       as first_center_network_code
        ,dt                              as dt
    from jms_dwd.dwd_wide_unsign_summary_waybill_dt
    where dt between date_sub('{{ execution_date | cst_ds }}',1) and '{{ execution_date | cst_ds }}'
    and ordersource_name in ('逍遥峰')
) wide on base.waybill_no = wide.waybill_no
left join jms_dim.dim_network_whole_massage dim
  on nvl(step.scan_site_code,wide.pick_network_code) = dim.code
left join jms_dim.dim_network_whole_massage dim_start
  on wide.first_center_network_code = dim_start.code
distribute by dt,abs(hash(waybill_no)) % 10;
